
package com.wd.base.thread;

import android.os.Handler;
import android.os.Looper;
import android.os.Process;

import androidx.annotation.NonNull;
import androidx.annotation.Nullable;

import com.wd.base.util.Logger;
import com.wd.base.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <线程池工具类>
 *
 * @author wangnaiwen
 * @version [V1.0.0.0, 2020/4/7]
 * @since V1.0.0.0
 */
public final class ThreadPoolUtils {

    private static final String TAG = "ThreadPoolUtil";

    /**
     * CPU 个数
     */
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 主线程 Handler
     */
    private static final Handler MAIN_HANDLER = new Handler(Looper.getMainLooper());

    /**
     * 一般优先级的线程池容量：和 CPU 一样多
     */
    private static final int NORMAL_POOL_SIZE = CPU_COUNT;

    /**
     * 高优先级线程池大小
     */
    private static final int EMERGENCY_POOL_SIZE = 2;

    /**
     * 低优先级线程池大小
     */
    private static final int BACKGROUND_POOL_SIZE = 2;

    /**
     * 高优先级线程池
     */
    private static final ThreadPoolExecutor EMERGENCY_THREAD_POOL =
            new ThreadPoolExecutor(EMERGENCY_POOL_SIZE, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue(),
                    new PriorityGroupedThreadFactory(Thread.MAX_PRIORITY, (String) null));;

    /**
     * 普通优先级线程池
     */
    private static final ThreadPoolExecutor NORMAL_THREAD_POOL =
            new ThreadPoolExecutor(NORMAL_POOL_SIZE, NORMAL_POOL_SIZE, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue(),
                    new PriorityGroupedThreadFactory(Thread.NORM_PRIORITY, (String) null));;

    /**
     * 低优先级线程池
     */
    private static final ThreadPoolExecutor BACKGROUND_THREAD_POOL = new ThreadPoolExecutor(BACKGROUND_POOL_SIZE, 2, 0L,
            TimeUnit.SECONDS, new LinkedBlockingQueue(), new PriorityGroupedThreadFactory(1, (String) null));

    /**
     * 等待线程池
     */
    private static final ScheduledThreadPoolExecutor SCHEDULED_THREAD_POOL =
            new ScheduledThreadPoolExecutor(NORMAL_POOL_SIZE);;

    /**
     * 对象锁
     */
    private static final Object GROUP_LOCK = new Object();

    /**
     * 普通的线程组池大小：1. （CPU个数 -1） 和 4 取最小值得到值 A 2. 值 A 和 2 取最大值，保证不低于两组
     */
    private static final int NORMAL_GROUP_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));

    /**
     * 线程组池的容器
     */
    private static final Map<String, ThreadPoolExecutor> GROUP_THREAD_POOLS = new HashMap<>();

    private ThreadPoolUtils() {
    }

    /**
     * 子线程执行:普通优先级
     *
     * @param runnable 运行接口
     * @return 取消接口
     */
    public static Cancelable submit(Runnable runnable) {
        return submitInner(runnable, Thread.NORM_PRIORITY, (String) null);
    }

    /**
     * 子线程执行：普通优先级
     *
     * @param groupName 指定的线程池
     * @param runnable 执行
     * @return 取消
     */
    public static Cancelable submitWithGroup(String groupName, Runnable runnable) {
        return submitInner(runnable, Thread.NORM_PRIORITY, groupName);
    }

    /**
     * 关闭线程组
     *
     * @param groupName 组名字
     */
    public static void shutdownGroup(String groupName) {
        ThreadPoolExecutor groupPool;
        synchronized (GROUP_LOCK) {
            groupPool = (ThreadPoolExecutor) GROUP_THREAD_POOLS.remove(groupName);
        }

        if (groupPool != null) {
            groupPool.shutdown();
            Logger.i(TAG, "group thread pool shutdown:" + groupName);
        }

    }

    /**
     * 子线程池执行：高优先级
     *
     * @param runnable 运行接口
     * @return 取消接口
     */
    public static Cancelable emergencySubmit(Runnable runnable) {
        return submitInner(runnable, Thread.MAX_PRIORITY, (String) null);
    }

    /**
     * 子线程池执行：高优先级
     *
     * @param runnable 运行接口
     * @return 取消接口
     */
    public static Cancelable backgroundSubmit(Runnable runnable) {
        return submitInner(runnable, Thread.MIN_PRIORITY, (String) null);
    }

    private static Cancelable submitInner(Runnable runnable, int priority, String groupName) {
        if (null == runnable) {
            Logger.e(TAG, "submit: Task is null, priority:" + PriorityGroupedThreadFactory.getPriorityName(priority)
                    + ", group:" + groupName);
            return SubmitCancelable.from((Future) null);
        } else {
            ThreadPoolExecutor actualPool = getThreadPoolInner(groupName, priority);
            BlockingQueue<Runnable> queue = actualPool.getQueue();
            int queueSizeBeforeSend = queue.size();
            long currTime = System.currentTimeMillis();
            PriorityRunnable actualRunnable = new PriorityRunnable(runnable, priority, queueSizeBeforeSend, currTime);

            try {
                Future<?> future = actualPool.submit(actualRunnable);
                return SubmitCancelable.from(future);
            } catch (RejectedExecutionException var10) {
                Logger.e(TAG, "submit: Task is rejected, priority:"
                        + PriorityGroupedThreadFactory.getPriorityName(priority) + ", group:" + groupName);
                return SubmitCancelable.from((Future) null);
            }
        }
    }

    /**
     * 延迟提交
     *
     * @param runnable 执行接口
     * @param delayTime 延迟时间
     * @param unit 时间单位
     * @return 取消接口
     */
    public static Cancelable schedule(Runnable runnable, long delayTime, TimeUnit unit) {
        Runnable intermediate = LifecycleRunnable.transfer(runnable);
        Runnable actualRunnable = AutoReleaseRunnable.transfer(intermediate);
        Future<?> future = SCHEDULED_THREAD_POOL.schedule(actualRunnable, delayTime, unit);
        return SubmitCancelable.from(future);
    }

    /**
     * 周期性执行线程
     *
     * @param runnable 执行接口
     * @param initialDelay 延迟时间
     * @param period 周期时间
     * @param unit 单位
     * @return 取消接口
     */
    public static Cancelable scheduleAtFixedRate(Runnable runnable, long initialDelay, long period, TimeUnit unit) {
        Runnable intermediate = LifecycleRunnable.transfer(runnable);
        Runnable actualRunnable = AutoReleaseRunnable.transfer(intermediate);
        Future<?> future = SCHEDULED_THREAD_POOL.scheduleAtFixedRate(actualRunnable, initialDelay, period, unit);
        return SubmitCancelable.from(future);
    }

    private static ThreadPoolExecutor getThreadPoolInner(String groupName, int priority) {
        if (!StringUtils.isEmpty(groupName)) {
            return getGroupedThreadPool(groupName, -1, -1);
        } else if (priority == Thread.MAX_PRIORITY) {
            return EMERGENCY_THREAD_POOL;
        } else {
            return priority == Thread.MIN_PRIORITY ? BACKGROUND_THREAD_POOL : NORMAL_THREAD_POOL;
        }
    }

    /**
     * 获得组线程池
     *
     * @param groupName 组名字
     * @param threadCount 线程数量
     * @param priority 线程的优先级
     * @return 线程池
     */
    public static ThreadPoolExecutor getGroupedThreadPool(@NonNull String groupName, int threadCount, int priority) {
        int actualThreadCount = threadCount <= 0 ? NORMAL_GROUP_POOL_SIZE : threadCount;
        int actualPriority = priority == -1 ? Thread.NORM_PRIORITY : priority;
        synchronized (GROUP_LOCK) {
            // 如果在缓存中拿不到线程池，则新建一个线程池，并且添加到缓存中
            ThreadPoolExecutor groupPool = (ThreadPoolExecutor) GROUP_THREAD_POOLS.get(groupName);
            if (groupPool == null) {
                groupPool = new ThreadPoolExecutor(actualThreadCount, actualThreadCount, 30L, TimeUnit.SECONDS,
                        new LinkedBlockingQueue(), new PriorityGroupedThreadFactory(actualPriority, groupName));
                groupPool.allowCoreThreadTimeOut(true);
                GROUP_THREAD_POOLS.put(groupName, groupPool);
            }

            return groupPool;
        }
    }

    /**
     * 主线程执行
     *
     * @param runnable 执行接口
     * @return 取消接口
     */
    public static Cancelable postToMain(Runnable runnable) {
        Runnable intermediate = LifecycleRunnable.transfer(runnable);
        Runnable actualRunnable = AutoReleaseRunnable.transfer(intermediate);
        MAIN_HANDLER.post(actualRunnable);
        return PostCancelable.from(MAIN_HANDLER, actualRunnable);
    }

    /**
     * 主线程延时执行
     *
     * @param runnable 执行接口
     * @param delayMillis 延时时间：毫秒
     * @return 取消接口
     */
    public static Cancelable postToMainDelay(Runnable runnable, long delayMillis) {
        Runnable intermediate = LifecycleRunnable.transfer(runnable);
        Runnable actualRunnable = AutoReleaseRunnable.transfer(intermediate);
        MAIN_HANDLER.postDelayed(actualRunnable, delayMillis);
        return PostCancelable.from(MAIN_HANDLER, actualRunnable);
    }

    /**
     * 线程工厂类
     */
    private static class PriorityGroupedThreadFactory implements ThreadFactory {
        /**
         * 线程数量
         */
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        /**
         * 命名前缀
         */
        private final String namePrefix;

        /**
         * 线程优先级
         */
        private final int threadPriority;

        PriorityGroupedThreadFactory(int priority, String groupName) {
            this.threadPriority = priority;
            String group = groupName == null ? "def" : StringUtils.str2LowerCase(groupName);
            this.namePrefix = "x-" + getPriorityName(priority) + "-" + group + "-";
        }

        static String getPriorityName(int priority) {
            if (priority == 10) {
                return "emrg";
            } else {
                return priority == 1 ? "back" : "norm";
            }
        }

        public Thread newThread(@NonNull Runnable r) {
            Thread t = new Thread(r, this.namePrefix + this.threadNumber.getAndIncrement());
            if (t.isDaemon()) {
                t.setDaemon(false);
            }

            if (this.threadPriority == 1) {
                t.setPriority(1);
            } else {
                t.setPriority(5);
            }

            return t;
        }
    }

    /**
     * 线程取消接口
     */
    private static final class SubmitCancelable implements Cancelable {
        private Future<?> futureTask;

        private SubmitCancelable(@Nullable Future<?> future) {
            this.futureTask = future;
        }

        static SubmitCancelable from(@Nullable Future<?> future) {
            return new SubmitCancelable(future);
        }

        public void cancel() {
            if (this.futureTask == null) {
                Logger.e(TAG, "future not set, cannot cancel");
            } else {
                this.futureTask.cancel(true);
            }
        }

        public boolean isCanceled() {
            if (this.futureTask == null) {
                Logger.e(TAG, "future not set, same as canceled");
                return false;
            } else {
                return this.futureTask.isCancelled();
            }
        }
    }

    /**
     * 带有优先级响应接口
     */
    private static class PriorityRunnable implements Runnable {
        private final int threadPriorityInner;

        private final Runnable runnableInner;

        private int queueSizeBeforeSend;

        private long queueTime;

        private static long WARNING_WAIT_THRESHOLD = 1000L;

        PriorityRunnable(Runnable runnable, int priority, int queueSizeBeforeSend, long queueTime) {
            Runnable intermediate = LifecycleRunnable.transfer(runnable);
            this.runnableInner = AutoReleaseRunnable.transfer(intermediate);
            this.threadPriorityInner = priority == 1 ? 10 : 0;
            this.queueSizeBeforeSend = queueSizeBeforeSend;
            this.queueTime = queueTime;
        }

        public void run() {
            if (this.runnableInner == null) {
                Logger.w(TAG, "inner runnable is null");
            } else {
                long currTime = System.currentTimeMillis();
                long waitTime = currTime - this.queueTime;
                if (waitTime > WARNING_WAIT_THRESHOLD) {
                    Thread currThread = Thread.currentThread();
                    Logger.w(TAG,
                            "wait too long in queue, wait time:" + waitTime + ", waitNum:" + this.queueSizeBeforeSend
                                    + ", tName:" + currThread.getName() + ", tPry:" + currThread.getPriority());
                }

                Process.setThreadPriority(this.threadPriorityInner);
                this.runnableInner.run();
            }
        }
    }
}